package com.crpc.core.discovery.Impl;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.crpc.common.constant.Constant;
import com.crpc.common.crpcEnum.CrpcExceptionEnums;
import com.crpc.common.exception.CrpcException;
import com.crpc.common.utils.NetWork.NetUtils;
import com.crpc.common.utils.zookeeper.ZookeeperNode;
import com.crpc.common.utils.zookeeper.ZookeeperUtils;
import com.crpc.core.CrpcBootStrap;
import com.crpc.core.ServiceConfig;
import com.crpc.core.discovery.AbstractRegistry;
import com.crpc.core.watcher.UpAndDownWatcher;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooKeeper;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author CLOUD
 * @version 1.0
 * @description:
 * @date 2023/9/8 19:18
 */
@Slf4j
public class ZookeeperRegistry extends AbstractRegistry {
    private ZooKeeper zooKeeper;

    public ZookeeperRegistry() {
        this.zooKeeper = ZookeeperUtils.createZooKeeper();
    }

    public ZookeeperRegistry(String connectString, int timeout) {
        this.zooKeeper = ZookeeperUtils.createZooKeeper(connectString, timeout);
    }


    @Override
    public void register(ServiceConfig<?> serviceConfig) {
//服务名称的节点
        //持久节点
        String parentNode = Constant.BASE_PROVIDERS_PATH + "/" + serviceConfig.getInterfaceProvider().getName();
        if (!ZookeeperUtils.exists(zooKeeper, parentNode, null)) {
            ZookeeperNode build = ZookeeperNode.builder().nodePath(parentNode).build();
            ZookeeperUtils.createNode(build, zooKeeper, null, CreateMode.PERSISTENT);
        }
        //临时节点
        //服务提供方的端口一般自己设定,默认8088,获取ip方法
        // 端口
        String providerHost;
        if (CrpcBootStrap.getInstance().getConfiguration().getProviderHost().equals("127.0.0.1")){
            providerHost=NetUtils.getIntranetIp();
        }else {
            providerHost=CrpcBootStrap.getInstance().getConfiguration().getProviderHost();
        }
        String node = parentNode + "/" + providerHost + ":" + CrpcBootStrap.getInstance().getConfiguration().getPort();
        if (!ZookeeperUtils.exists(zooKeeper, node, null)) {
            ZookeeperNode build = ZookeeperNode.builder().nodePath(node).build();
            ZookeeperUtils.createNode(build, zooKeeper, null, CreateMode.EPHEMERAL);
        }
        log.info("服务：{}，已经被发布", serviceConfig.getInterfaceProvider().getName());
    }

    /**
     * @description: 注册中心的目的是拉取合适的服务列表
     * @author wzy
     * @date 2023/9/30 21:23
     */
    @Override
    public List<InetSocketAddress> lookup(String serviceName) {
        //拼接路径,找到节点
        String serviceNode = Constant.BASE_PROVIDERS_PATH + "/" + serviceName;
        //从zk获取子节点
        UpAndDownWatcher upAndDownWatcher = new UpAndDownWatcher();
        List<String> children = ZookeeperUtils.getChildren(zooKeeper, serviceNode, upAndDownWatcher);
        //获取所有服务可用列表
        if (ObjectUtil.isNull(children) || children.isEmpty()) {
            return new ArrayList<>();
        }
        List<InetSocketAddress> collect = children.stream().map(ipString -> {
            String[] ipAndPort = ipString.split(":");
            String ip = ipAndPort[0];
            Integer port = Integer.valueOf(ipAndPort[1]);
//            Integer port = Integer.getInteger(ipAndPort[1]);
            return new InetSocketAddress(ip, port);
        }).collect(Collectors.toList());
        if (collect.isEmpty()) {
            throw new CrpcException(CrpcExceptionEnums.NETTY_ERROR);
        }
        return collect;
    }
}
